package org.nbict.iot.trash;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.mongodb.common.mapper.SimpleMongoMapper;
import org.apache.storm.topology.TopologyBuilder;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.trash.mogo.MongoInsertBolt;

import java.util.regex.Pattern;

/**
 * Created by songseven on 18/6/19.
 */
@Deprecated
public class LogTopology {

    public static void main(String[] args) throws InvalidTopologyException,
            AuthorizationException, AlreadyAliveException {

        TopologyBuilder builder = new TopologyBuilder();

        KafkaSpoutConfig.Builder<String, String> kafkaBuilder = new
                KafkaSpoutConfig.Builder("192.168.1.83:9092", StringDeserializer
                .class, StringDeserializer.class, Pattern.compile("transit-.*-simple")); //transit-5vvokv 需要监听多个

        kafkaBuilder.setGroupId("iot-car-group-log");
        KafkaSpoutConfig build = kafkaBuilder.build();
        KafkaSpout kafkaSpout = new KafkaSpout(build);

        //step1. 报文读取流
        builder.setSpout("kafkaSpout", kafkaSpout, 1);

        //step2. 报文日志解析
        builder.setBolt("log4mBolt", new Log4MongoBolt(), 3)
                .shuffleGrouping("kafkaSpout");

        //step3. 输出
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String adminname = PropertyUtil.getProperty("mongo.admin.name");
        String url = PropertyUtil.getProperty("mongo.url");

        MongoMapper mapper = new SimpleMongoMapper().withFields("platform_id", "car_vin","action", "timestamp", "info");
        MongoInsertBolt insertBolt = new MongoInsertBolt(url, username, password, adminname, database, "car_history", mapper);
        builder.setBolt("save", insertBolt, 2).shuffleGrouping("log4mBolt");

        Config config = new Config();

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, builder
                    .createTopology());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-log", config, builder
                    .createTopology());
        }
    }
}
